package com.example.yckjbigdataservice.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class HdfsFileScanner {

    private final FileSystem fs;
    private final Set<String> excludeDirs;
    private final List<String> suffixFilters;
    private final int maxDepth;

    public HdfsFileScanner(FileSystem fs, Set<String> excludeDirs, List<String> suffixFilters, int maxDepth) throws IOException {
        this.fs = fs;
        this.excludeDirs = excludeDirs != null ? excludeDirs : new HashSet<>();
        this.suffixFilters = suffixFilters != null ? suffixFilters : new ArrayList<>();
        this.maxDepth = maxDepth;
    }

    public List<Path> scan(Path startPath) throws IOException {
        List<Path> result = new ArrayList<>();
        scanInternal(startPath, result, 0);
        return result;
    }

    private void scanInternal(Path path, List<Path> result, int depth) throws IOException {
        if (depth > maxDepth) {
            return;
        }

        FileStatus[] statuses = fs.listStatus(path);
        if (statuses == null) return;

        for (FileStatus status : statuses) {
            Path currentPath = status.getPath();
            String name = currentPath.getName();

            if (status.isDirectory()) {
                //if(excludeDirs.parallelStream().noneMatch(dir -> currentPath.toString().contains(dir))) {
                if (!excludeDirs.contains(name)) {
                    scanInternal(currentPath, result, depth + 1);
                }
            } else {
                if (matchesSuffix(currentPath.getName())) {
                    result.add(currentPath);
                }
            }
        }
    }

    private boolean matchesSuffix(String fileName) {
        if (suffixFilters.isEmpty()) return true;
        for (String suffix : suffixFilters) {
            if (fileName.endsWith(suffix)) {
                return true;
            }
        }
        return false;
    }

    public void close() throws IOException {
        fs.close();
    }
}
